/**
 * 
 */
package org.datagear.management.util;

import java.lang.reflect.Field;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import org.datagear.management.domain.Schema;
import org.springframework.beans.BeanUtils;
import org.springframework.dao.QueryTimeoutException;

import com.google.api.client.util.Lists;
import com.google.cloud.ServiceOptions;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQueryOptions;
import com.google.cloud.bigquery.FieldValue;
import com.google.cloud.bigquery.FieldValueList;
import com.google.cloud.bigquery.QueryJobConfiguration;
import com.google.cloud.bigquery.TableResult;
import com.google.common.collect.Maps;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.digest.MD5;
import cn.hutool.json.JSONUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * lc-dc
 * 
 * @author kome
 * @date 2020年11月25日 下午2:50:10
 */
@Slf4j
public class BigQueryClient {
	private final BigQuery bigquery;

	private static PluralCache<String, Object> cache = new PluralCache<>(3000, 300);

	public BigQueryClient(BigQuery bigQuery) {
		this.bigquery = bigQuery;
	}

	/**
	 * 2. 创建QueryDemo类调用查询方法
	 * 三,设置身份验证(linux)
	 * 
	 * @description 查询方法
	 * @author rhc
	 */
	public void runQuery(String query) throws InterruptedException {
		QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(query).build();
		// Print the results.
		for (FieldValueList row : bigquery.query(queryConfig).iterateAll()) {
			for (FieldValue val : row) {
				System.out.println(String.format("%s,%s ", val.toString(), val.getStringValue()));
			}
		}
	}

	public <T> List<T> runQuery(String query, List<String> fields, Class<T> clazz) {
		List<T> ls = Lists.newArrayList();

		QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(query).build();
		try {
			for (FieldValueList row : bigquery.query(queryConfig).iterateAll()) {

				T t = clazz.newInstance();
				for (int i = 0; i < row.size(); i++) {
//					 BeanUtils.invokeSetter(t, fields.get(i), row.get(i).getValue());
					BeanUtil.setFieldValue(t, fields.get(i), row.get(i).getValue());
				}

				ls.add(t);
			}
		} catch (Exception e) {
			log.warn("{}", query, e);
		}
		return ls;
	}

	public static BigQuery getBigQuery(Schema schema) {
		// 测试bigQuery
		// -Djava.net.useSystemProxies=true -Dhttps.proxyHost=127.0.0.1 -Dhttps.proxyPort=1080
		if (StrUtil.isNotEmpty(schema.getUser())) {
			System.setProperty("java.net.useSystemProxies", "true");
			System.setProperty("https.proxyHost", schema.getUser());
			System.setProperty("https.proxyPort", schema.getPassword());
		}

		BigQueryClient.setEnv(ServiceOptions.CREDENTIAL_ENV_NAME, schema.getUrl());

		// 创建BigQuery客户端对象
		return BigQueryOptions.getDefaultInstance().getService();
	}

	public static TableResult doQuery(Schema schema, String sql) {
		// -Djava.net.useSystemProxies=true -Dhttps.proxyHost=127.0.0.1 -Dhttps.proxyPort=1080
		BigQuery bigquery = BigQueryClient.getBigQuery(schema);
		log.info("BigQuery config: {}", System.getenv(ServiceOptions.CREDENTIAL_ENV_NAME));
		QueryJobConfiguration queryConfig = QueryJobConfiguration.newBuilder(sql).build();
		try {
			TableResult rs = tryCache(sql);
			rs = bigquery.query(queryConfig);
			cache(sql, rs);
			return rs;
		} catch (Exception e) {
			throw new QueryTimeoutException("BigQuery: " + sql);
		}
	}

	private static String key(String sql) {
		return MD5.create().digestHex(sql);
	}

	private static TableResult tryCache(String sql) {
		String key = key(sql);
		Object rs = cache.get(key);
		if (rs != null) {
			return (TableResult) rs;
		}
		return null;
	}

	private static void cache(String sql, TableResult rs) {
		if (rs != null) {
			String key = key(sql);
			cache.put(key, rs);
		}
	}

	public static void setEnv(String k, String v) {
		Map<String, String> map = Maps.newHashMap();
		map.put(k, v);
		setEnv(map);
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	public static void setEnv(Map<String, String> newenv) {
		try {
			Class<?> processEnvironmentClass = Class.forName("java.lang.ProcessEnvironment");
			Field theEnvironmentField = processEnvironmentClass.getDeclaredField("theEnvironment");
			theEnvironmentField.setAccessible(true);
			Map<String, String> env = (Map<String, String>) theEnvironmentField.get(null);
			env.putAll(newenv);
			Field theCaseInsensitiveEnvironmentField = processEnvironmentClass.getDeclaredField("theCaseInsensitiveEnvironment");
			theCaseInsensitiveEnvironmentField.setAccessible(true);
			Map<String, String> cienv = (Map<String, String>) theCaseInsensitiveEnvironmentField.get(null);
			cienv.putAll(newenv);
		} catch (Exception e) {
			try {
				Class[] classes = Collections.class.getDeclaredClasses();
				Map<String, String> env = System.getenv();
				for (Class cl : classes) {
					if ("java.util.Collections$UnmodifiableMap".equals(cl.getName())) {
						Field field = cl.getDeclaredField("m");
						field.setAccessible(true);
						Object obj = field.get(env);
						Map<String, String> map = (Map<String, String>) obj;
						map.clear();
						map.putAll(newenv);
					}
				}
			} catch (Exception ex) {
				log.warn("setEnv failed!", ex);
			}
		}
	}

	@Data
	public static class User {
		private String hour;
		private int server;
		private String plateform;
		private int users;
	}

	public static void main(String[] args) throws InterruptedException {
		// -Djava.net.useSystemProxies=true -Dhttps.proxyHost=127.0.0.1 -Dhttps.proxyPort=1080
		System.setProperty("java.net.useSystemProxies", "true");
		System.setProperty("https.proxyHost", "localhost");
		System.setProperty("https.proxyPort", "1080");

		setEnv(ServiceOptions.CREDENTIAL_ENV_NAME, "d:/home/nova-empire-168011-2fac63a18e09.json");
		System.out.println(System.getenv(ServiceOptions.CREDENTIAL_ENV_NAME));

		// 创建BigQuery客户端对象
		BigQuery bigquery = BigQueryOptions.getDefaultInstance().getService();
		BigQueryClient client = new BigQueryClient(bigquery);
		// 执行查询
		String query = "SELECT DATETIME_TRUNC(datetime(ts,'+08'),HOUR) AS hour,  server,   platform,   level,   COUNT(DISTINCT uid ) AS users "
				+ "FROM   `nova-empire-168011.appsflyer_installs_report.server_data` "
				+ "WHERE   cmd = 'REGIST'   AND DATE(ts,'+08') = CURRENT_DATE('+08')  GROUP BY   1,   2,   3,   4";
		List<User> list = client.runQuery(query, CollUtil.newArrayList("hour", "server", "plateform", "level", "users"), User.class);
		System.out.println(JSONUtil.toJsonStr(list));
	}
}